package com.qiaofang.housewarehouse.dao.elasticsearch;


import com.alibaba.fastjson.JSONObject;
import com.qiaofang.housewarehouse.dao.domain.InitESIndex;
import com.qiaofang.housewarehouse.dao.utils.ConfigUtils;
import com.qiaofang.housewarehouse.dao.utils.DateUtils;
import com.qiaofang.housewarehouse.stub.constants.TableConstants;
import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
import org.elasticsearch.action.bulk.byscroll.BulkByScrollResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchPhaseExecutionException;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHitField;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;
import java.net.InetAddress;
import java.util.*;

/**
 * Created by Administrator on 2017/7/27.
 */
public class EsClient {

    private static Logger logger = LoggerFactory.getLogger(EsClient.class);

    public static TransportClient client = null;
    private static String clusterName = ConfigUtils.getEsClusterName();//设置集群名称
    private static boolean sniff = ConfigUtils.isEsSniff();// 是否自动检测变化节点
    private static String pingTimeout = ConfigUtils.getEsPingTimeout();// 等待来自节点的ping响应的时间。默认为5 s
    private static boolean ignoreClusterName = ConfigUtils.isEsIgnoreClusterName();//是否忽略集群名称
    private static String nodesSamplerInterval = ConfigUtils.getEsNodesSamplerInterval();// 对列出和连接的节点进行采样的频率。默认为5 s
    private static String IP = ConfigUtils.getEsIP();
    private static int PORT = ConfigUtils.getEsPORT();  //端口

    static {
        try {
            logger.info("ES初始化配置开始");
            Settings settings = Settings.builder().put("cluster.name", clusterName).put("client.transport.sniff", sniff)
                    .put("client.transport.ignore_cluster_name", ignoreClusterName)
                    .put("client.transport.ping_timeout", pingTimeout)
                    .put("client.transport.nodes_sampler_interval", nodesSamplerInterval).build();
            client = new PreBuiltTransportClient(settings);
            if (IP != null) {
                if (IP.indexOf(",") != -1) {
                    for (String ip : IP.split(",")) {
                        client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip), Integer.valueOf(PORT)));
                    }
                } else {
                    client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(IP), Integer.valueOf(PORT)));
                }
            }
          //  initES();
            logger.info("ES 集群{}初始化配置====SUCCESS", clusterName);
        } catch (Exception e) {
            logger.error("ES初始化配置失败:{}", e.getMessage(), e);
        }
    }

    public static void initES() throws Exception {
        InitESIndex initESIndex = new InitESIndex();
        initESIndex.setId("init");
        initESIndex.setCity("init");
        initESIndex.setCreateTime(DateUtils.formatDate(new Date(), DateUtils.YYYY_MM_DD_HH_MM));
        EsClient.addIndex(TableConstants.INDEX_SALE_HOUSE, TableConstants.INDEX_TYPE_SALE_HOUSE, initESIndex);
        EsClient.addIndex(TableConstants.INDEX_RENTAL_HOUSE, TableConstants.INDEX_TYPE_RENTAL_HOUSE, initESIndex);
    }

    /**
     * 根据queryBuilder查询
     *
     * @param index        索引
     * @param type         类型
     * @param queryBuilder 查询语句
     * @param size         返回多少条
     * @return
     */
    public static List<JSONObject> searchSearchResponseByQueryBuilder(String index, String type, QueryBuilder queryBuilder, Integer size) {
        int pageSize = 10000;
        if (null != size) {
            pageSize = size;
        }
        logger.info("QueryBuilder查询语句 : {}", queryBuilder.toString());
        SearchResponse searchResponse = client.prepareSearch(index)
                .setTypes(type)
                .setQuery(queryBuilder)
                .setFrom(0)
                .setSize(pageSize)
                .get();
        List<JSONObject> jsonObjectList = new ArrayList<>();
        for (SearchHit searchHit : searchResponse.getHits().getHits()) {
            if (searchHit.getSourceAsMap() == null) {
                return null;
            }
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("id", searchHit.getId());
            Map<String, SearchHitField> fields = searchHit.getFields();
            for (String key : fields.keySet()) {
                jsonObject.put(key.replace("_", ""), fields.get(key).getValue());
            }
            Map<String, Object> source = searchHit.getSource();
            for (String key : source.keySet()) {
                jsonObject.put(key, source.get(key));
            }
            jsonObjectList.add(jsonObject);
        }
        return jsonObjectList;
    }

    /**
     * 添加索引
     *
     * @param index
     * @param type
     * @param object
     * @return
     * @throws IllegalAccessException
     */
    public static String addIndex(String index, String type, Object object) throws IllegalAccessException {
        logger.info("操作es集群{}", clusterName);
        HashMap<String, Object> hashMap = new HashMap<>();
        Class cls = object.getClass();
        Field[] fields = cls.getDeclaredFields();
        String id = null;
        for (Field f : fields) {
            f.setAccessible(true);
            if (f.getName().equals("id")) {
                id = (String) f.get(object);
            } else {
                hashMap.put(f.getName(), f.get(object) == null ? "" : f.get(object));
            }
        }
        IndexResponse response = client.prepareIndex(index, type, id).setSource(hashMap).execute().actionGet();
        return response.getId();
    }

    //查询数据
    public static SearchResponse queryData(String index, String type, QueryBuilder queryBuilder, int offset, int rows) throws Exception {
        logger.info("操作es集群{}", clusterName);
        SortBuilder sortBuilder = SortBuilders.fieldSort("updatePublishTime").order(SortOrder.DESC);
        SearchResponse searchResponse = null;
        try {
            searchResponse = client.prepareSearch(index).setTypes(type).setQuery(queryBuilder).addSort(sortBuilder).setFrom(offset).setSize(rows).execute().actionGet();
        } catch (SearchPhaseExecutionException e) {
            if ((e.getCause().getMessage().equals("No mapping found for [updatePublishTime] in order to sort on"))) {
                logger.info("es数据为空{}", e.getCause().getMessage());
            } else {
                throw e;
            }
        }
        return searchResponse;
    }

    //查询不分页数据
    public static SearchResponse queryDataNoPage(String index, String type, QueryBuilder queryBuilder) throws Exception {
        logger.info("操作es集群{}", clusterName);
        SortBuilder sortBuilder = SortBuilders.fieldSort("createTime").order(SortOrder.DESC);
        SearchResponse searchResponse = null;
        try {
            searchResponse = client.prepareSearch(index).setTypes(type).setQuery(queryBuilder).addSort(sortBuilder).setFrom(0).setSize(10000).execute().actionGet();
        } catch (SearchPhaseExecutionException e) {
            if ((e.getCause().getMessage().equals("No mapping found for [createTime] in order to sort on"))) {
                logger.info("es数据为空{}", e.getCause().getMessage());
            } else {
                throw e;
            }
        }
        return searchResponse;
    }

    //查询不带条件的数据
    public static SearchResponse queryDataNoCondition(String index, String type, QueryBuilder queryBuilder) throws Exception {
        logger.info("操作es集群{}", clusterName);
        SearchResponse searchResponse = null;
        try {
            searchResponse = client.prepareSearch(index).setTypes(type).setQuery(queryBuilder).setSize(0).execute().actionGet();
        } catch (SearchPhaseExecutionException e) {
                throw e;
        }
        return searchResponse;
    }

    //getindex
    public static GetResponse getIndexById(String index, String type, String id) throws Exception {
        logger.info("操作es集群{}", clusterName);
        return client.prepareGet(index, type, id).get();
    }

    //查询index
    public static List<String> queryIndexs() {
        try {
            GetIndexResponse response = client.admin().indices().prepareGetIndex().get();
            String[] indices = response.getIndices();
            List<String> projectNames = new ArrayList<>();
            //String projectName = new String();
            for (int i = 0, l = indices.length; i < l; i++) {
                if (indices[i].matches("ms-.*")) {
                    projectNames.add(indices[i]);
                }
            }
            Collections.sort(projectNames);
            return projectNames;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    //删除索引中部分数据
    public static long deleteData(String index, String field, Long datetime) {
        try {
            BulkByScrollResponse response = DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
                    .filter(QueryBuilders.rangeQuery(field).lte(datetime)).source(index).get();
            return response.getDeleted();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return -1L;
    }
}
